package com.fwmagic.flink.projectcase.queryactivitycase;

import com.fwmagic.flink.projectcase.queryactivitycase.bean.ActivityBean;
import com.fwmagic.flink.projectcase.queryactivitycase.map.GPSToActivityBeanMapFunction;
import com.fwmagic.flink.projectcase.queryactivitycase.utils.FlinkUtilsV1;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 需求：查询高德地图API，关联地理位置信息
 *
 * 准备发送到kafka的数据：
 * u001,A1,2019-09-02 10:10:11,1,115.908923,39.267291
 * u002,A2,2019-08-03 10:11:11,2,123.818817,41.312458
 * u003,A3,2019-06-08 10:11:11,3,121.26757,37.49794
 *
 * 期望打印结果：
 * u001,A1,2019-09-02 10:10:11,1,河北省
 * u002,A2,2019-08-03 10:11:11,2,辽宁省
 * u003,A3,2019-06-08 10:11:11,3,山东省
 */
public class QueryProvinceByGPSApplication {
    public static void main(String[] args) throws Exception{

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //获取kafkaSource
        DataStream<String> kafkaSource = FlinkUtilsV1.createKafkaSource(args, new SimpleStringSchema(),env);

        //map操作数据
        SingleOutputStreamOperator<ActivityBean> beans = kafkaSource.map(new GPSToActivityBeanMapFunction());

        beans.print();

        env.execute("QueryProvinceByGPSApplication");

    }
}
